package com.smsc.headend.task.engine.service.impl;

import cn.hutool.core.date.SystemClock;
import com.smsc.headend.common.utils.ProtoBufferUtil;
import com.smsc.headend.module.task.consts.TaskKafkaTopic;
import com.smsc.headend.module.task.dto.AtomicTask;
import com.smsc.headend.module.task.dto.ConnectorTask;
import com.smsc.headend.module.task.dto.Task;
import com.smsc.headend.module.task.proto.ConnectorTaskProto;
import com.smsc.headend.task.engine.service.ConnectorService;
import io.micrometer.core.instrument.Counter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

@Service
@Slf4j
public class ConnectorServiceImpl implements ConnectorService {
    @Autowired
    KafkaTemplate kafkaTemplate;

    @Autowired
    @Qualifier("taskCONOutbound")
    Counter conTaskOutboundCounter;

    @Override
    public void sendPacket(Task task, AtomicTask atomicTask, Integer roleId, Boolean isReturnPacket, byte[]... bytes) {
        log.debug("SendTask: taskNo={}, atomicTaskNo={}, taskType={}, roleId={}, isReturn={}", task.getTaskNo(), atomicTask.getAtomicTaskNo(), task.getTaskType(), roleId, isReturnPacket);
        ConnectorTask taskObject = new ConnectorTask();
        taskObject.setAtomicTaskNo(atomicTask.getAtomicTaskNo());
        taskObject.setTaskNo(task.getTaskNo());
        taskObject.setTerminalId(task.getComId());
        taskObject.setMeterId(task.getMeterId());
        taskObject.setPacketBytes(bytes);
        taskObject.setRoleId(roleId);
        taskObject.setReturnPacket(isReturnPacket);
        taskObject.setSendTime(SystemClock.now());
//        log.info(JSONUtil.toJsonStr(taskObject));
        if (task.getPriority() <= Task.Priority.ABOVENORMAL.getPriorityCode()) {
            taskObject.setFlag(taskObject.getFlag() | 0x08);
        }
        ConnectorTaskProto.ConnectorTask proto = ProtoBufferUtil.toProtoBuffer(taskObject, ConnectorTaskProto.ConnectorTask.class);
        byte[] res = proto.toByteArray();
        conTaskOutboundCounter.increment();
        kafkaTemplate.send(TaskKafkaTopic.TASK_SEND_TO_CON, res);
    }

    /**
     * flag为2 发送GPRS AT指令报文
     * @param task
     * @param atomicTask
     * @param roleId
     * @param isReturnPacket
     * @param bytes
     */
    @Override
    public void sendAtCmdPacket(Task task, AtomicTask atomicTask, Integer roleId, Boolean isReturnPacket, Boolean isSetAction, byte[]... bytes) {
        log.debug("SendTask: taskNo={}, atomicTaskNo={}, taskType={}, roleId={}, isReturn={}", task.getTaskNo(), atomicTask.getAtomicTaskNo(), task.getTaskType(), roleId, isReturnPacket);
        ConnectorTask taskObject = new ConnectorTask();
        taskObject.setAtomicTaskNo(atomicTask.getAtomicTaskNo());
        taskObject.setTaskNo(task.getTaskNo());
        taskObject.setTerminalId(task.getComId());
        taskObject.setMeterId(task.getMeterId());
        taskObject.setPacketBytes(bytes);
        taskObject.setFlag(isSetAction?6:2); // AT指令任务get为2 set为6
        taskObject.setRoleId(roleId);
        taskObject.setReturnPacket(isReturnPacket);
        taskObject.setSendTime(SystemClock.now());
        if (task.getPriority() <= Task.Priority.ABOVENORMAL.getPriorityCode()) {
            taskObject.setFlag(taskObject.getFlag() | 0x08);
        }
        ConnectorTaskProto.ConnectorTask proto = ProtoBufferUtil.toProtoBuffer(taskObject, ConnectorTaskProto.ConnectorTask.class);
        byte[] res = proto.toByteArray();
        conTaskOutboundCounter.increment();
        kafkaTemplate.send(TaskKafkaTopic.TASK_SEND_TO_CON, res);
    }

    /**
     * flag is 1 ,is recreate aa for key transfer
     *
     * @param task
     * @param atomicTask
     * @param roleId
     * @param isReturnPacket
     */
    @Override
    public void sendForReCreateAA(Task task, AtomicTask atomicTask, Integer roleId, Boolean isReturnPacket) {
        log.debug("SendTask: taskNo={}, atomicTaskNo={}, taskType={}, roleId={}, isReturn={}", task.getTaskNo(), atomicTask.getAtomicTaskNo(), task.getTaskType(), roleId, isReturnPacket);
        ConnectorTask taskObject = new ConnectorTask();
        taskObject.setAtomicTaskNo(atomicTask.getAtomicTaskNo());
        taskObject.setTaskNo(task.getTaskNo());
        taskObject.setTerminalId(task.getComId());
        taskObject.setMeterId(task.getMeterId());
        taskObject.setPacketBytes(null);
        taskObject.setFlag(1);
        taskObject.setRoleId(roleId);
        taskObject.setReturnPacket(isReturnPacket);
        taskObject.setSendTime(SystemClock.now());
        if (task.getPriority() <= Task.Priority.ABOVENORMAL.getPriorityCode()) {
            taskObject.setFlag(taskObject.getFlag() | 0x08);
        }
        ConnectorTaskProto.ConnectorTask proto = ProtoBufferUtil.toProtoBuffer(taskObject, ConnectorTaskProto.ConnectorTask.class);
        byte[] res = proto.toByteArray();
        conTaskOutboundCounter.increment();
        kafkaTemplate.send(TaskKafkaTopic.TASK_SEND_TO_CON, res);
    }

}
